/**
 * Software License, Version 1.0
 *
 * Copyright 2003 The Trustees of Indiana University.  All rights reserved.
 *
 *
 *Redistribution and use in source and binary forms, with or without
 *modification, are permitted provided that the following conditions are met:
 *
 *1) All redistributions of source code must retain the above copyright notice,
 * the list of authors in the original source code, this list of conditions and
 * the disclaimer listed in this license;
 *2) All redistributions in binary form must reproduce the above copyright
 * notice, this list of conditions and the disclaimer listed in this license in
 * the documentation and/or other materials provided with the distribution;
 *3) Any documentation included with all redistributions must include the
 * following acknowledgement:
 *
 *"This product includes software developed by the Community Grids Lab. For
 * further information contact the Community Grids Lab at
 * http://communitygrids.iu.edu/."
 *
 * Alternatively, this acknowledgement may appear in the software itself, and
 * wherever such third-party acknowledgments normally appear.
 *
 *4) The name Indiana University or Community Grids Lab or NaradaBrokering,
 * shall not be used to endorse or promote products derived from this software
 * without prior written permission from Indiana University.  For written
 * permission, please contact the Advanced Research and Technology Institute
 * ("ARTI") at 351 West 10th Street, Indianapolis, Indiana 46202.
 *5) Products derived from this software may not be called NaradaBrokering,
 * nor may Indiana University or Community Grids Lab or NaradaBrokering appear
 * in their name, without prior written permission of ARTI.
 *
 *
 * Indiana University provides no reassurances that the source code provided
 * does not infringe the patent or any other intellectual property rights of
 * any other entity.  Indiana University disclaims any liability to any
 * recipient for claims brought by any other entity based on infringement of
 * intellectual property rights or otherwise.
 *
 *LICENSEE UNDERSTANDS THAT SOFTWARE IS PROVIDED "AS IS" FOR WHICH NO
 *WARRANTIES AS TO CAPABILITIES OR ACCURACY ARE MADE. INDIANA UNIVERSITY GIVES
 *NO WARRANTIES AND MAKES NO REPRESENTATION THAT SOFTWARE IS FREE OF
 *INFRINGEMENT OF THIRD PARTY PATENT, COPYRIGHT, OR OTHER PROPRIETARY RIGHTS.
 *INDIANA UNIVERSITY MAKES NO WARRANTIES THAT SOFTWARE IS FREE FROM "BUGS",
 *"VIRUSES", "TROJAN HORSES", "TRAP DOORS", "WORMS", OR OTHER HARMFUL CODE.
 *LICENSEE ASSUMES THE ENTIRE RISK AS TO THE PERFORMANCE OF SOFTWARE AND/OR
 *ASSOCIATED MATERIALS, AND TO THE PERFORMANCE AND VALIDITY OF INFORMATION
 *GENERATED USING SOFTWARE.
 */
package cgl.narada.samples.nbNative.tds;

/**
 * This class is adapted from cgl.narada.samples.rtp.NBRTPConnector. This class
 * does not use RTPEvent. Instead it uses native NBEvent to transmit and receive
 * the stream.
 * 
 * @author Hasan Bulut - hbulut@indiana.edu Community Grids Lab., Indiana
 *         University
 * @version 1.0
 */

import java.io.IOException;
import java.util.Properties;

import javax.media.protocol.ContentDescriptor;
import javax.media.protocol.PushSourceStream;
import javax.media.protocol.SourceTransferHandler;
import javax.media.rtp.OutputDataStream;
import javax.media.rtp.RTPConnector;

import cgl.narada.event.EventProperties;
import cgl.narada.event.NBEvent;
import cgl.narada.event.NBEventException;
import cgl.narada.event.TemplateInfo;
import cgl.narada.event.TemplateProfileAndSynopsisTypes;
import cgl.narada.matching.Profile;
import cgl.narada.service.ServiceException;
import cgl.narada.service.client.ClientService;
import cgl.narada.service.client.EventConsumer;
import cgl.narada.service.client.EventProducer;
import cgl.narada.service.client.NBEventListener;
import cgl.narada.service.client.NBRecoveryListener;
import cgl.narada.service.client.NBRecoveryNotification;
import cgl.narada.service.client.SessionService;
import cgl.narada.service.qos.ConsumerConstraints;
import cgl.narada.service.qos.ProducerConstraints;
import cgl.narada.service.replay.ReplayEvent;
import cgl.narada.service.replay.ReplayRequest;
import cgl.narada.service.replay.ReplayResponse;
import cgl.narada.service.replay.ReplayServiceListener;
import cgl.narada.service.timer.HRClock;
import cgl.narada.transport.TransportHandler;

public class NBMediaConnector implements RTPConnector, NBEventListener,
    NBRecoveryListener, ReplayServiceListener {

    private String ip; // broker ip

    private int brokerPort; // broker port

    private int localPort;
    
    private int receiveBufferSize;

    private int sendBufferSize;

    private NBInputStream dataInputStream = null;

    private NBInputStream controlInputStream = null;

    private NBOutputStream dataOutputStream = null;

    private NBOutputStream controlOutputStream = null;

    protected TransportHandler transportHandler; // remove

    private String dataTopic = "Media/Data";

    private String controlTopic = "Media/Control";

    private int entityID;
    
    private ClientService clientService;

    private EventConsumer receiver;

    private EventProducer transmitter;

    private int templateID = 3142;

    private EventProperties evtProp;

    private Integer evtID;

    private String moduleName = "NBMediaConnector: ";

    private ProducerConstraints producerConstraints;

    private boolean debug = true;

    private boolean withTDS = false;

    private boolean publishEnabled = true;

    private int tracker = 0;

    public NBMediaConnector(String ip, int brkPort, int entityID,
        int topicNumber) {
        this.ip = ip;
        this.brokerPort = brkPort;
        dataTopic = dataTopic + "/" + topicNumber;
        controlTopic = controlTopic + "/" + (topicNumber + 1);
        this.entityID = entityID;
        evtID = new Integer(entityID);
        try {
            SessionService
                .setServiceConfigurationLocation("C:/NaradaBrokering/config/ServiceConfiguration.txt");
            clientService = SessionService.getClientService(entityID);
        } catch (Exception ex) {
            ex.printStackTrace();
        }
    }

    public NBMediaConnector(String ip, int brkPort, int entityID) {
        this.ip = ip;
        this.brokerPort = brkPort;
        this.entityID = entityID;
        evtID = new Integer(entityID);
        try {
            String configLocation = "C:/NB-Rel/config/ServiceConfiguration.txt";
            SessionService.setServiceConfigurationLocation(configLocation);
            clientService = SessionService.getClientService(entityID);
        } catch (Exception ex) {
            ex.printStackTrace();
        }
    }

    public void initializeBrokerCommunications(Properties props, String commType)
        throws ServiceException {
        clientService.initializeBrokerCommunications(props, commType);
        receiver = clientService.createEventConsumer(this);
    }

    public void initializeReceiver() throws ServiceException {
        withTDS = false;
        Profile dataProfile = clientService
            .createProfile(TemplateProfileAndSynopsisTypes.STRING, dataTopic);
        Profile controlProfile = clientService
            .createProfile(TemplateProfileAndSynopsisTypes.STRING, controlTopic);

        ConsumerConstraints constraints = receiver
            .createConsumerConstraints(dataProfile);
        // constraints.setReceiveReliably(templateID);
        // constraints.setReceiveInTimeOrder();
        // constraints.setReceiveAfterTimeSpacing();
        receiver.subscribeTo(dataProfile, constraints);
        // receiver.recover(templateID, this);
        receiver.subscribeTo(controlProfile);
    }

    public void initializeReceiverWithTDS() throws ServiceException {
        withTDS = true;
        Profile dataProfile = clientService
            .createProfile(TemplateProfileAndSynopsisTypes.STRING, dataTopic);
        Profile controlProfile = clientService
            .createProfile(TemplateProfileAndSynopsisTypes.STRING, controlTopic);

        ConsumerConstraints constraints = receiver
            .createConsumerConstraints(dataProfile);
        // constraints.setReceiveReliably(templateID);
        constraints.setReceiveInTimeOrder();
        constraints.setReceiveAfterTimeSpacing();
        receiver.subscribeTo(dataProfile, constraints);
        // receiver.recover(templateID, this);
        receiver.subscribeTo(controlProfile);

    }

    public void initiateReplay(long start, long stop) throws ServiceException {
        ReplayRequest replayRequest = clientService
            .createReplayRequest(templateID, start, stop);
        receiver.initiateReplay(replayRequest, this);
    }

    public void onReplay(ReplayEvent replayEvent) {
        System.out.println(moduleName + "Received Replay Event");
        onEvent(replayEvent.getNBEvent());
    }

    public void onReplayResponse(ReplayResponse replayResponse) {
        System.out.println("\n\n\n\n" + moduleName + replayResponse + "\n\n");
    }

    public void initializeTransmitter() throws ServiceException {
        transmitter = clientService.createEventProducer();
        transmitter.setSuppressRedistributionToSource(true);
        transmitter.generateEventIdentifier(true);
        transmitter.setTemplateId(templateID);
        transmitter.setHighResolutionTimestamp(true);
        transmitter.setDisableTimestamp(false);

        TemplateInfo templateInfo = clientService
            .createTemplateInfo(templateID,
                                TemplateProfileAndSynopsisTypes.STRING,
                                dataTopic);
        producerConstraints = transmitter
            .createProducerConstraints(templateInfo);
        producerConstraints.setSendReliably();
        long recoveryId = transmitter.recover(templateID, this);
        System.out.println(moduleName + "Assigned recovery id = [" + recoveryId
            + "] \n\n");
    }

    public void onEvent(NBEvent nbEvent) {
        String synopsis = (String) nbEvent.getContentSynopsis();
        if (debug) {
            System.out.println(moduleName + "Received NBEvent ");
        }

        if (!withTDS) {
            HRClock.getClock().sleep(5);
        }

        if (synopsis.equals(dataTopic)) {
            if (dataInputStream != null) {
                dataInputStream.onMessage(nbEvent);
            } else {
                System.err.println(moduleName
                    + "Data received but dataInputStream is null.");
            }
            return;
        } else if (synopsis.equals(controlTopic)) {
            if (controlInputStream != null) {
                controlInputStream.onMessage(nbEvent);
            } else {
                System.err.println(moduleName
                    + "Data received but controlInputStream is null.");
            }
            return;
        }
    }

    /**
     * Upon completion of the attempt to recover, this method is invoked on the
     * listener that was registered with the
     */
    public void onRecovery(NBRecoveryNotification recoveryNotification) {
        System.out.println(moduleName + recoveryNotification);
    }

    /** Enable or diable publishing of streams */
    public void setPublishEnabled(boolean toPublish) {
        publishEnabled = toPublish;
    }

    public void publishEvent(byte[] data, int offset, int length, String tpcName)
        throws ServiceException, NBEventException {
        if (transmitter == null) {
            System.out.println("null transmitter");
            System.exit(0);
        }

        if (publishEnabled) {

            NBEvent nbEvent = transmitter
                .generateEvent(TemplateProfileAndSynopsisTypes.STRING, tpcName,
                               data);

            if (tpcName.equals(dataTopic)) {
                tracker++;
                transmitter.publishEvent(nbEvent, producerConstraints);
                System.out.println(moduleName + "Published (" + tracker
                    + ") events");
            } else {
                transmitter.publishEvent(nbEvent);
            }/* end if dataTopic */

        } /* end if (publishEnabled) */

    }

    public boolean isClient() {
        return true;
    }

    public void close() {
        try {
            closeBrokerConnection();
        } catch (ServiceException ex) {
            ex.printStackTrace();
        }
    }

    public void closeBrokerConnection() throws ServiceException {
        clientService.closeBrokerConnection();
        clientService.terminateServices();
    }

    /**
     * Returns an input stream to receive the RTP data.
     */
    public PushSourceStream getDataInputStream() {
        if (dataInputStream == null) {
            dataInputStream = new NBInputStream(dataTopic);
        }
        return dataInputStream;
    }

    /**
     * Returns an output stream to send the RTP data.
     */
    public OutputDataStream getDataOutputStream() {
        if (dataOutputStream == null) {
            dataOutputStream = new NBOutputStream(dataTopic);
        }
        return dataOutputStream;
    }

    /**
     * Returns an input stream to receive the RTCP data.
     */
    public PushSourceStream getControlInputStream() {
        if (controlInputStream == null) {
            controlInputStream = new NBInputStream(controlTopic);
        }
        return controlInputStream;
    }

    /**
     * Returns an output stream to send the RTCP data.
     */
    public OutputDataStream getControlOutputStream() {
        if (controlOutputStream == null) {
            controlOutputStream = new NBOutputStream(controlTopic);
        }

        return controlOutputStream;
    }

    /**
     * Set the receive buffer size of the RTP data channel. This is only a hint
     * to the implementation. The actual implementation may not be able to do
     * anything to this.
     */
    public void setReceiveBufferSize(int size) throws IOException {
        receiveBufferSize = size;
    }

    /**
     * Get the receive buffer size set on the RTP data channel. Return -1 if the
     * receive buffer size is not applicable for the implementation.
     */
    public int getReceiveBufferSize() {
        return receiveBufferSize;
    }

    /**
     * Set the send buffer size of the RTP data channel. This is only a hint to
     * the implementation. The actual implementation may not be able to do
     * anything to this.
     */
    public void setSendBufferSize(int size) throws IOException {
        sendBufferSize = size;
    }

    /**
     * Get the send buffer size set on the RTP data channel. Return -1 if the
     * send buffer size is not applicable for the implementation.
     */
    public int getSendBufferSize() {
        return sendBufferSize;
    }

    /**
     * Return the RTCP bandwidth fraction. This value is used to initialize the
     * RTPManager. Check RTPManager for more detauls. Return -1 to use the
     * default values.
     */
    public double getRTCPBandwidthFraction() {
        return -1;
    }

    /**
     * Return the RTCP sender bandwidth fraction. This value is used to
     * initialize the RTPManager. Check RTPManager for more detauls. Return -1
     * to use the default values.
     */
    public double getRTCPSenderBandwidthFraction() {
        return -1;
    }

    class NBOutputStream implements OutputDataStream {
        private String topicName;

        boolean isDataStream;

        public NBOutputStream(String tpcName) {
            this.topicName = tpcName;
            if (topicName.equalsIgnoreCase(dataTopic)) {
                isDataStream = true;
            } else {
                isDataStream = false;
            }
        }

        public int write(byte[] data, int offset, int len) {
            // send the package
            try {
                publishEvent(data, offset, len, topicName);
            } catch (Exception e) {
                e.printStackTrace();
                return -1;
            }
            return len;
        }
    } // end of class NBOutputStream

    class NBInputStream extends Thread implements PushSourceStream {
        SourceTransferHandler sth = null;

        NBEvent message;

        private String topicName;

        Object synchObject = new Object();

        EventProperties evtProp;

        int off = 0;

        int len = 0;

        public NBInputStream(String tpcName) {
            this.topicName = tpcName;
        }

        public int read(byte[] buffer, int offset, int length) {
            synchronized (synchObject) {
                off = 0;
                len = message.getContentPayloadSize();
                System.arraycopy(message.getContentPayload(), off, buffer,
                                 offset, len);
            }
            return message.getContentPayloadSize();
        }

        public void onMessage(NBEvent message) {
            synchronized (synchObject) {
                this.message = message;
            }
            if (sth != null) {
                sth.transferData(this);
            } else {
                System.out.println(this.getClass().getName()
                    + ": sth is null .......................");
            }
        }

        public int getMinimumTransferSize() {
            return 2 * 1024; // twice the MTU size, just to be safe.
        }

        public synchronized void setTransferHandler(SourceTransferHandler sth) {
            this.sth = sth;
        }

        // Not applicable.
        public ContentDescriptor getContentDescriptor() {
            return null;
        }

        // Not applicable.
        public long getContentLength() {
            return LENGTH_UNKNOWN;
        }

        // Not applicable.
        public boolean endOfStream() {
            return false;
        }

        // Not applicable.
        public Object[] getControls() {
            return new Object[0];
        }

        // Not applicable.
        public Object getControl(String type) {
            return null;
        }
    } // end of class NBInputStream
}
